package org.bling.star.server.connect;

import org.bling.star.common.command.systems.PingCommand;
import org.bling.star.common.command.systems.SystemCommand;
import org.bling.star.common.connect.ConnectClient;
import org.bling.star.common.connect.ConnectStatus;
import org.bling.star.server.connect.node.InstanceConnection;
import io.netty.channel.Channel;
import org.slf4j.Logger;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static org.bling.star.common.enums.SystemCommandType.CONNECT_PING;

/**
 * @Describe: 所有外部连接channel管理
 * @Date: 2024/7/24
 * @Author: szz
 */

public class GlobalConnectManager {

    private final Channel fatherChannel;
    private final Set<InstanceConnection> connections = new HashSet<>();
    private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
    private final Set<InstanceConnection> prepareDisconnections = new HashSet<>();
    private static final Logger log = org.slf4j.LoggerFactory.getLogger(GlobalConnectManager.class);
    private static GlobalConnectManager globalConnectManager;

    public static GlobalConnectManager getInstance() {
        if (globalConnectManager == null) {
            log.error("GlobalConnectManager is null");
        }
        return globalConnectManager;
    }

    public GlobalConnectManager(Channel fatherChannel) {
        this.fatherChannel = fatherChannel;
        globalConnectManager = this;
    }

    public ConnectClient getFrom(Channel channel) {
        ConnectClient streamConnection = connections.stream().filter(connection -> connection.channel().equals(channel)).findFirst().orElse(null);
        if (streamConnection != null) {
            return streamConnection;
        }else {
            InstanceConnection instanceConnection = new InstanceConnection(channel);
            addConnection(instanceConnection);
            return instanceConnection;
        }
    }

    public void addConnection(InstanceConnection connection) {
        connections.add(connection);
        log.info("add connection:{},connections:{}" , connection.getChannel().toString(),connections.size());
    }

    public void refresh() {
        PingCommand pingCommand = new PingCommand();
        pingCommand.setCommandType(CONNECT_PING);

        executorService.scheduleAtFixedRate(() -> {
            connections.forEach(connection -> {
                    if (connection.status() == ConnectStatus.MISSING) {
                        if (prepareDisconnections.contains(connection.channel())) {
                            prepareDisconnections.remove(connection.channel());
                            connection.close();
                        }
                    }
                    connection.sendFuture(pingCommand).addListener(future -> {
                        log.info("ping:{}", connection.getChannel().toString());
                        if (!future.isSuccess()) {
                            if (prepareDisconnections.contains(connection)) {
                                prepareDisconnections.remove(connection);
                                connections.remove(connection);
                                connection.close();
                            } else {
                                connection.setMissing();
                                prepareDisconnections.add(connection);
                            }
                            log.info("fail,connections.size():{}", connections.size());
                        }else {
                            log.info("error");
                            connection.alive();
                            prepareDisconnections.remove(connection.channel());
                        }
                    });
            });
        }, 2, 20, TimeUnit.SECONDS);
    }

    public void syncListener() {
        try {
            fatherChannel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public void close() {
        executorService.shutdown();
        prepareDisconnections.clear();
        broadcast(new SystemCommand());
        connections.forEach(connection -> {
            connection.close();
        });
        connections.clear();
        fatherChannel.close();
    }
    public void broadcast(SystemCommand systemCommand) {
        connections.forEach(connection -> {
            connection.sendFuture(systemCommand);
        });
    }
}
